package com.miya.demo.support.kafka.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * kafka消息发送handler
 * 这个注解代表这个类开启Springboot事务，因为我们在Kafka的配置文件开启了Kafka事务，不然会报错
 *
 * @author CaiXiaowei
 * @date 2023/6/1
 */
@Component
@Slf4j
@Transactional(rollbackFor = RuntimeException.class)
public class KafkaSendHandler {

    public static final String TOPIC = "topic.demo";

    private final KafkaTemplate<Object, Object> kafkaTemplate;

    public KafkaSendHandler(KafkaTemplate<Object, Object> kafkaTemplate, KafkaSendResultHandler kafkaSendResultHandler) {
        this.kafkaTemplate = kafkaTemplate;
        //回调方法、异常处理
        this.kafkaTemplate.setProducerListener(kafkaSendResultHandler);
    }

    /**
     * 同步发送
     *
     * @param message 消息
     * @throws ExecutionException   执行异常
     * @throws InterruptedException 中断异常
     * @throws TimeoutException     超时异常
     */
    public void sendSync(String message) throws ExecutionException, InterruptedException, TimeoutException {
        kafkaTemplate.send(TOPIC, message).get(1, TimeUnit.SECONDS);
    }

    /**
     * 发送生产记录
     *
     * @param message 消息
     * @throws ExecutionException   执行异常
     * @throws InterruptedException 中断异常
     * @throws TimeoutException     超时异常
     */
    public void sendProducerRecord(String message) {
        ProducerRecord<Object, Object> producerRecord = new ProducerRecord<>(TOPIC, message);
        kafkaTemplate.send(producerRecord);
    }

    /**
     * 发送消息
     *
     * @param message 消息
     * @throws ExecutionException   执行异常
     * @throws InterruptedException 中断异常
     * @throws TimeoutException     超时异常
     */
    public void sendMessage(String message) {
        //使用Message发送消息
        Map<String, Object> map = new HashMap<>(8);
        map.put(KafkaHeaders.TOPIC, TOPIC);
        map.put(KafkaHeaders.PARTITION_ID, 0);
        map.put(KafkaHeaders.MESSAGE_KEY, 0);
        GenericMessage<Object> genericMessage = new GenericMessage<>(message, new MessageHeaders(map));
        kafkaTemplate.send(genericMessage);

    }
}
